// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_MINOR_COMPACT_CHECKER_HPP_
#define _XBASE_MINOR_COMPACT_CHECKER_HPP_

#include <vector>

#include <boost/foreach.hpp>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/noncopyable.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#include <xthread/guard.hpp>
#include <xthread/fast_lock.hpp>

#include "global.hpp"
#include "table_manager.hpp"
#include "tablet_server_configuration.hpp"
#include "tablet.hpp"

namespace xbase {

	class minor_compact_checker
		: private boost::noncopyable {
	public:
		minor_compact_checker()
			: m_stop(false)
			, m_duration(global::the_conf->minor_compact_check_interval) {}

		void run() {
			LOG(INFO, "minor_compact_checker::run checker started");
			m_deadline_time = boost::get_system_time();
			m_deadline_time += m_duration;
			while(! m_stop)
			{
				tablet::pointer tablet = global::get_minor_compact_tablet(m_deadline_time);
				if(m_stop)
				{
					break;
				}
				
				if(! tablet) // timeout
				{
					LOG(INFO, "waiting timeout, check each tablet");
					check_tablet();
					// update timer
					m_deadline_time += m_duration;
				}
				else
				{
					LOG(INFO, "got notified, tablet = " << tablet->tablet_name());
					// 将该子表锁定，这将阻止子表下线
					//Guard<ReadWriteLock::ReadLock> guard(tablet->lock().read_lock());
					boost::shared_lock<boost::shared_mutex> guard(tablet->lock());
					if(! tablet->is_online()) // double check, not online
					{
						LOG(DEBUG, "minor_compact_checker::run tablet "
						    << tablet->tablet_name()
						    << " is not online, skip it");
						continue;
					}
					tablet->start_minor_compact();
					bool need_major_compact = false;
					const bool ok = tablet->do_minor_compact(need_major_compact);
					LOG_IF((!ok),
					       ERROR,
					       "minor_compact_checker::run minor compact error on tablet "
					       << tablet->tablet_name());
					if(need_major_compact)
					{
						tablet->notify_major_compact();
					} 
					else
					{
						tablet->complete_minor_compact();
					}
				}
			}
		}

		void stop() {
			m_stop = true;
			global::stop_minor_compact();
		}

	protected:
		void check_tablet() const {
			LOG(TRACE, "minor_compact_checker::check_tablet start checking");
			std::vector<table::pointer> list;
			global::the_table_manager->list_table(list);
			BOOST_FOREACH(const table::pointer &elem, list)
			{
				check_table(elem);
			}
			LOG(TRACE, "minor_compact_checker::check_tablet checking done");
		}

		void check_table(table::pointer table) const {
			std::vector<tablet::pointer> list;
			table->list_tablet(list);
			BOOST_FOREACH(const tablet::pointer &elem, list)
			{
				if(elem->is_online())
				{
					check_tablet(elem);
				}
			}
		}
		
		void check_tablet(tablet::pointer tablet) const {
			// 将该子表锁定，这将阻止子表下线
			//Guard<ReadWriteLock::ReadLock> guard(tablet->lock().read_lock());
			boost::shared_lock<boost::shared_mutex> guard(tablet->lock());
			if(! tablet->is_online()) // double check, not online
			{
				LOG(DEBUG, "minor_compact_checker::check tablet "
				    << tablet->tablet_name()
				    << " is not online, skip it");
				return;
			}
			tablet->check_compact();
		}

	private:
		bool m_stop;
		const boost::posix_time::seconds m_duration;
		boost::system_time m_deadline_time;
	};
	
} // namespace xbase

#endif	// _XBASE_MINOR_COMPACT_CHECKER_HPP_
